/*
 * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */
package com.i360day.invoker;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitUtils;

import java.io.IOException;
import java.util.Collections;
import java.util.function.Function;

/**
 * @author liju.z
 * @date 2024/3/27 20:48
 */
public class SimpleRabbitmqTemplate {
    private final ConnectionFactory connectionFactory;
    private Connection connection;
    private Channel channel;

    public SimpleRabbitmqTemplate(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    public Connection createConnection() {
        if (connection != null && connection.isOpen()) {
            return connection;
        }
        RabbitUtils.closeConnection(connection);
        return connection = connectionFactory.createConnection();
    }

    public Channel createChannel(boolean flag) {
        if (channel != null && channel.isOpen()) {
            return channel;
        }
        RabbitUtils.closeChannel(channel);
        return channel = createConnection().createChannel(flag);
    }

    /**
     * 执行
     *
     * @param function
     * @param <R>
     * @return
     */
    public <R> R execute(Function<Channel, R> function) {
        try{
            return function.apply(createChannel(false));
        }finally {
            RabbitUtils.closeChannel(channel);
            RabbitUtils.closeConnection(connection);
        }
    }

    /**
     * 注册队列绑定
     *
     * @param exchange
     * @param queue
     * @param binding
     */
    public void queueBinding(Exchange exchange, Queue queue, Binding binding) {
        try {
            exchangeDeclare(exchange);
            //队列注册，并绑定
            if (queueDeclare(queue)) {
                queueBinding(binding);
            }
        } catch (Exception ex) {
            throw new IllegalStateException(String.format("build queue[%s] to exchange[%s] routing [%s]", binding.getDestination(), binding.getExchange(), binding.getRoutingKey()), ex);
        }
    }

    /**
     * 绑定队列
     *
     * @param binding
     */
    public void queueBinding(Binding binding) {
        Channel channel = createChannel(false);
        try {
            channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(), binding.getArguments());
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    }

    /**
     * 注册交换机
     *
     * @param exchange
     */
    public void exchangeDeclare(Exchange exchange) {
        Channel channel = createChannel(false);
        try {
            channel.exchangeDeclare(exchange.getName(), exchange.getType(), exchange.isDurable(), exchange.isAutoDelete(), exchange.isInternal(), exchange.getArguments());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 注册队列
     *
     * @param queue
     * @return
     */
    public boolean queueDeclare(Queue queue) {
        Channel channel = createChannel(false);
        try {
            if (queueDeclarePassive(queue)) {
                channel.queueDeclare(queue.getName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete(), queue.getArguments());
            }
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    /**
     * 删除
     *
     * @param queue
     * @return
     */
    public boolean queueDelete(String queue) {
        Channel channel = createChannel(false);
        try {
            channel.queueDelete(queue);
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    /**
     * 队列是否存在
     *
     * @param queue
     * @return
     */
    public boolean queueDeclarePassive(Queue queue) {
        Channel channel = createChannel(false);
        try {
            AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive(queue.getName());
            return false;
        } catch (IOException e) {
            return true;
        }
    }

    /**
     * 监听消息回复
     *
     * @param replyTo
     * @param autoAck
     * @param callback
     * @return
     * @throws IOException
     */
    public DefaultMessageReply basicConsumeReply(String replyTo, boolean autoAck, DefaultMessageReply callback) throws IOException {
        queueDeclare(new Queue(replyTo, true, true, false, Collections.emptyMap()));
        basicConsume(replyTo, autoAck, callback);
        return callback;
    }

    /**
     * 监听队列
     *
     * @param queue
     * @param autoAck
     * @param callback
     * @throws IOException
     */
    public void basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException {
        Channel channel = createChannel(false);
        channel.basicConsume(queue, autoAck, callback);
    }

    /**
     * 发送队列消息
     *
     * @param exchange
     * @param routingKey
     * @param props
     * @param body
     * @throws IOException
     */
    public void basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body) throws IOException {
        Channel channel = createChannel(false);
        channel.basicPublish(exchange, routingKey, props, body);
    }
}
